package com.ruoyi.rabbitmq.service;

import com.ruoyi.rabbitmq.config.RabbitMQConfig;
import com.ruoyi.rabbitmq.domain.MessageInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

import com.rabbitmq.client.Channel;
import java.io.IOException;

/**
 * 消息消费者服务
 * 
 * @author ruoyi
 */
@Service
@ConditionalOnProperty(name = "spring.rabbitmq.enabled", havingValue = "true", matchIfMissing = true)
public class MessageConsumer
{
    private static final Logger log = LoggerFactory.getLogger(MessageConsumer.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MessagePublisher messagePublisher;

    /**
     * 监听直连队列
     */
    @RabbitListener(queues = RabbitMQConfig.DIRECT_QUEUE)
    public void handleDirectMessage(MessageInfo messageInfo, Channel channel,
                                   @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
    {
        log.info("收到直连队列消息: {}", messageInfo);
        processMessage(messageInfo, channel, deliveryTag, "DIRECT");
    }

    /**
     * 监听主题队列1
     */
    @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE_1)
    public void handleTopicMessage1(MessageInfo messageInfo, Channel channel, 
                                   @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
    {
        log.info("收到主题队列1消息: {}", messageInfo);
        processMessage(messageInfo, channel, deliveryTag, "TOPIC_1");
    }

    /**
     * 监听主题队列2
     */
    @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE_2)
    public void handleTopicMessage2(MessageInfo messageInfo, Channel channel, 
                                   @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
    {
        log.info("收到主题队列2消息: {}", messageInfo);
        processMessage(messageInfo, channel, deliveryTag, "TOPIC_2");
    }

    /**
     * 监听扇形队列1
     */
    @RabbitListener(queues = RabbitMQConfig.FANOUT_QUEUE_1)
    public void handleFanoutMessage1(MessageInfo messageInfo, Channel channel, 
                                    @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
    {
        log.info("收到扇形队列1消息: {}", messageInfo);
        processMessage(messageInfo, channel, deliveryTag, "FANOUT_1");
    }

    /**
     * 监听扇形队列2
     */
    @RabbitListener(queues = RabbitMQConfig.FANOUT_QUEUE_2)
    public void handleFanoutMessage2(MessageInfo messageInfo, Channel channel, 
                                    @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
    {
        log.info("收到扇形队列2消息: {}", messageInfo);
        processMessage(messageInfo, channel, deliveryTag, "FANOUT_2");
    }

    /**
     * 监听死信队列
     */
    @RabbitListener(queues = RabbitMQConfig.DEAD_LETTER_QUEUE)
    public void handleDeadLetterMessage(MessageInfo messageInfo, Channel channel, 
                                       @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
    {
        log.warn("收到死信队列消息: {}", messageInfo);
        processDeadLetterMessage(messageInfo, channel, deliveryTag);
    }

    /**
     * 监听重试队列
     */
    @RabbitListener(queues = RabbitMQConfig.RETRY_QUEUE)
    public void handleRetryMessage(MessageInfo messageInfo, Channel channel, 
                                  @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
    {
        log.info("收到重试队列消息: {}", messageInfo);
        processRetryMessage(messageInfo, channel, deliveryTag);
    }

    /**
     * 处理消息的核心方法
     */
    private void processMessage(MessageInfo messageInfo, Channel channel, long deliveryTag, String queueType)
    {
        try
        {
            // 执行业务逻辑
            boolean success = executeBusinessLogic(messageInfo, queueType);
            
            if (success)
            {
                // 业务处理成功，确认消息
                channel.basicAck(deliveryTag, false);
                log.info("消息处理成功，已确认，消息ID: {}", messageInfo.getMessageId());
            }
            else
            {
                // 业务处理失败，根据重试次数决定是否重试
                handleMessageFailure(messageInfo, channel, deliveryTag);
            }
        }
        catch (Exception e)
        {
            log.error("消息处理异常，消息ID: {}", messageInfo.getMessageId(), e);
            handleMessageFailure(messageInfo, channel, deliveryTag);
        }
    }

    /**
     * 处理死信消息
     */
    private void processDeadLetterMessage(MessageInfo messageInfo, Channel channel, long deliveryTag)
    {
        try
        {
            // 记录死信消息日志
            log.error("死信消息详情 - 消息ID: {}, 内容: {}, 错误信息: {}", 
                     messageInfo.getMessageId(), messageInfo.getContent(), messageInfo.getErrorMessage());
            
            // 可以在这里实现死信消息的特殊处理逻辑
            // 比如发送告警、记录到数据库等
            
            // 确认死信消息
            channel.basicAck(deliveryTag, false);
        }
        catch (Exception e)
        {
            log.error("处理死信消息异常", e);
            try
            {
                channel.basicNack(deliveryTag, false, false);
            }
            catch (IOException ex)
            {
                log.error("拒绝死信消息异常", ex);
            }
        }
    }

    /**
     * 处理重试消息
     */
    private void processRetryMessage(MessageInfo messageInfo, Channel channel, long deliveryTag)
    {
        try
        {
            // 重试消息直接重新发送到原队列
            messagePublisher.sendMessage(messageInfo.getContent(), 
                                       messageInfo.getExchange(), 
                                       messageInfo.getRoutingKey(), 
                                       messageInfo.getMessageType());
            
            // 确认重试消息
            channel.basicAck(deliveryTag, false);
            log.info("重试消息已重新发送，消息ID: {}", messageInfo.getMessageId());
        }
        catch (Exception e)
        {
            log.error("处理重试消息异常，消息ID: {}", messageInfo.getMessageId(), e);
            try
            {
                channel.basicNack(deliveryTag, false, false);
            }
            catch (IOException ex)
            {
                log.error("拒绝重试消息异常", ex);
            }
        }
    }

    /**
     * 处理消息失败
     */
    private void handleMessageFailure(MessageInfo messageInfo, Channel channel, long deliveryTag)
    {
        try
        {
            messageInfo.incrementRetryCount();
            
            if (messageInfo.isMaxRetryExceeded())
            {
                // 超过最大重试次数，拒绝消息，进入死信队列
                channel.basicNack(deliveryTag, false, false);
                log.error("消息处理失败，超过最大重试次数，进入死信队列，消息ID: {}", messageInfo.getMessageId());
            }
            else
            {
                // 未超过最大重试次数，发送到重试队列
                long delayMillis = calculateRetryDelay(messageInfo.getRetryCount());
                messagePublisher.sendRetryMessage(messageInfo, delayMillis);
                
                // 确认原消息
                channel.basicAck(deliveryTag, false);
                log.info("消息处理失败，已发送到重试队列，消息ID: {}, 重试次数: {}, 延迟时间: {}ms", 
                        messageInfo.getMessageId(), messageInfo.getRetryCount(), delayMillis);
            }
        }
        catch (Exception e)
        {
            log.error("处理消息失败异常，消息ID: {}", messageInfo.getMessageId(), e);
            try
            {
                channel.basicNack(deliveryTag, false, false);
            }
            catch (IOException ex)
            {
                log.error("拒绝消息异常", ex);
            }
        }
    }

    /**
     * 执行业务逻辑
     */
    private boolean executeBusinessLogic(MessageInfo messageInfo, String queueType)
    {
        try
        {
            // 这里实现具体的业务逻辑
            log.info("执行业务逻辑，队列类型: {}, 消息内容: {}", queueType, messageInfo.getContent());
            
            // 模拟业务处理
            // 可以根据消息类型和内容执行不同的业务逻辑
            switch (queueType)
            {
                case "DIRECT":
                    return processDirectMessage(messageInfo);
                case "TOPIC_1":
                case "TOPIC_2":
                    return processTopicMessage(messageInfo);
                case "FANOUT_1":
                case "FANOUT_2":
                    return processFanoutMessage(messageInfo);
                default:
                    return true;
            }
        }
        catch (Exception e)
        {
            log.error("执行业务逻辑异常，消息ID: {}", messageInfo.getMessageId(), e);
            messageInfo.setErrorMessage(e.getMessage());
            return false;
        }
    }

    /**
     * 处理直连消息
     */
    private boolean processDirectMessage(MessageInfo messageInfo)
    {
        // 实现直连消息的具体业务逻辑
        log.info("处理直连消息: {}", messageInfo.getContent());
        return true;
    }

    /**
     * 处理主题消息
     */
    private boolean processTopicMessage(MessageInfo messageInfo)
    {
        // 实现主题消息的具体业务逻辑
        log.info("处理主题消息: {}", messageInfo.getContent());
        return true;
    }

    /**
     * 处理扇形消息
     */
    private boolean processFanoutMessage(MessageInfo messageInfo)
    {
        // 实现扇形消息的具体业务逻辑
        log.info("处理扇形消息: {}", messageInfo.getContent());
        return true;
    }

    /**
     * 计算重试延迟时间（指数退避算法）
     */
    private long calculateRetryDelay(int retryCount)
    {
        // 基础延迟时间1秒，每次重试延迟时间翻倍，最大延迟时间30秒
        long baseDelay = 1000L;
        long maxDelay = 30000L;
        long delay = baseDelay * (long) Math.pow(2, retryCount - 1);
        return Math.min(delay, maxDelay);
    }
} 